package com.to8to.kitt.esb.client;

import java.lang.reflect.Constructor;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.lang.StringUtils;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.to8to.kitt.Constants;
import com.to8to.kitt.client.ThriftClientConfig;
import com.to8to.kitt.esb.BusServer;
import com.to8to.kitt.esb.ZK;
import com.to8to.kitt.strategy.RandomStrategy;
import com.to8to.kitt.utils.LogUtils;

public class ThriftClientPoolFactory {

	public static Logger logger = LoggerFactory.getLogger(ThriftClientPoolFactory.class);
	
	public static Map<String, ThriftClientPool> ClientMap = new HashMap<String, ThriftClientPool>();
	
	public static Map<Class<?>, Object> serviceMap = new HashMap<Class<?>, Object>();
	
	public static ESBAddressMonitor addresses = new ESBAddressMonitor();
	
	static
	{
		addresses.start(ZK.ZK_CONFIG.servicePath  + "/" + BusServer.class.getName());
	}
	
	public static Object getClient(Class<?> service) 
	{
		synchronized (service) 
		{
			Object ifaceProxy = null;
			if (serviceMap.containsKey(service))
				ifaceProxy = serviceMap.get(service);
			else 
			{
				ThriftClientInvocationHandler handler = new ThriftClientInvocationHandler(service.getName());
				
				Class<?> iface = null;
				
				Class<?>[] classes = service.getDeclaredClasses();
				
				for (Class<?> clz : classes) {
					if (clz.getSimpleName().equals("Iface")) {
						iface = clz;
						break;	
					}
				}

				if(iface != null)
				{
					ifaceProxy = Proxy.newProxyInstance(ThriftClientPoolFactory.class.getClassLoader(), new Class<?>[] { iface }, handler);
					
					serviceMap.put(service, ifaceProxy);
				}
			}
			
			return ifaceProxy;
		}
	}
	
	public static TServiceClient createClient(String service, String host, int port, int timeoutms) throws Exception{
		try {

			logger.info("create thrift client ... start!!! service:{}, address:{}, timeout:{}", service, host + ":" + port, timeoutms);

			String clientClassName = service + "$Client";
			Class<?> cls = Class.forName(clientClassName);
			Constructor<?> constructor = cls.getConstructor(TProtocol.class);

			TTransport transport = new TFramedTransport(new TSocket(host, port, timeoutms), Constants.MAX_DATA_LENGTH);
			transport.open();

			TProtocol protocol = new TBinaryProtocol(transport);
			TServiceClient client = (TServiceClient) constructor.newInstance(protocol);

			logger.info("create thrift client ... success!!!");

			return client;

		} catch (Exception e) {
			logger.info("create thrift client ... failed!!! host:{}, port:{}, timeoutms:{} \n{}"
					, host, port, timeoutms, LogUtils.logStackTrace(e));
			throw e;
		}
	}
	
	
	public static void ensureAdd(String instance, String host, int port)
	{
		synchronized(ThriftClientPoolFactory.ClientMap)
		{
			for(Entry<String, ThriftClientPool> entry : ClientMap.entrySet())
			{
				String key = entry.getKey();
				String service = StringUtils.substringBefore(key, "_");
				ThriftClientPool pool = entry.getValue();
				if(!pool.clients.containsKey(instance))
				{
					TServiceClient client = null;
					try {
						client = createClient(service, host, port, Constants.DEFAULT_CLIENT_TIMEOUT);
					} catch (Exception e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					if(client != null)
						pool.add(instance, client);
				}
			}
		}
	}
	
	public static void remove(String instance)
	{
		synchronized(ThriftClientPoolFactory.ClientMap)
		{
			for(ThriftClientPool pool : ThriftClientPoolFactory.ClientMap.values())
				pool.remove(instance);
		}
	}
	
	public static Entry<String, TServiceClient> getThreadLocalClient(String service) {
		
		long id = Thread.currentThread().getId();
		
		String key = service + "_" + id;
		
		ThriftClientPool pool = ClientMap.get(key);
		
		if(pool == null || pool.clients.isEmpty())
		{
			logger.info("initialize thrift connection pool ... start!");
			pool = new ThriftClientPool(new RandomStrategy<Entry<String, TServiceClient>>());
			for(ThriftClientConfig config : addresses.ESBAddresses.values())
			{
				TServiceClient client = null;
				try {
					client = createClient(service, config.host, config.port, config.timeoutms);
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				if(client != null)
					pool.add(config.id, client);
			}
			if(!pool.clients.isEmpty())
				ClientMap.put(key, pool);
			
			logger.info("initialize thrift connection pool ... finish!");
			
			return pool.get();
		}
		else 
		{
			return pool.get();
		}
	}
}
